package com.study.flink.demo;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010;
import org.apache.flink.util.Collector;

import com.study.flink.demo.TestDemo.TaskDTO;
import com.study.flink.entity.PeterEntity;
import com.study.flink.source.PeterSource;

import lombok.extern.slf4j.Slf4j;
import scala.tools.jline_embedded.internal.Log;

/**
 * 实验：<br/>
 * 1. 这是一个自己产生数据并排序的实验。
 * 2. 水印、时间窗口和事件时间的关系。
 * 3. 两个source的操作（一个作为开关使用）。
 * 4. 尝试side output操作
 * @projectName: flink-demo
 * @author: peter
 * @description:
 * @date: 2019年8月7日 下午9:11:08
 * @version:
 */
@Slf4j
public class PeterDemo {
    public void run(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        
        DataStream<PeterEntity> wordStream = env.addSource(new PeterSource());
        wordStream.keyBy(1).window(TumblingEventTimeWindows.of(Time.minutes(5)))
            .process(new LogMap()).writeAsText("/data/bigdata/forflink/ticketspricing."
                + new SimpleDateFormat("YYMMddHHmmss").format(new Date()));
        wordStream.keyBy(1).flatMap(null);
        env.execute();
    }
    
    private void kafkaConsumer() {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", "localhost:9092");
        properties.setProperty("zookeeper.connect", "localhost:2181");
        properties.setProperty("group.id", "test");
        FlinkKafkaConsumer010<String> consumer = new FlinkKafkaConsumer010<String>("peter", new SimpleStringSchema(), properties);
        
        consumer.setStartFromGroupOffsets();
        
//        RegistryAvroDeserializationSchema<TaskDTO> registrySchema = RegistryAvroDeserializationSchema

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.addSource(consumer);
    }
    
    public void eventCEP() {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        Properties properties = new Properties();
        properties.put("bootstrap.servers", "localhost:9092");
        properties.setProperty("zookeeper.connect", "localhost:2181");
        properties.setProperty("group.id", "test");
        FlinkKafkaConsumer010<String> consumer = new FlinkKafkaConsumer010<String>("peter", new SimpleStringSchema(), properties);

        DataStream<String> inputStream = env.addSource(consumer);
        Pattern<String, String> pattern = Pattern.<String>begin("start").where(
                new SimpleCondition<String>() {
                    private static final long serialVersionUID = 1L;

                    @Override
                    public boolean filter(String value) throws Exception {
                        return value.startsWith("pe");
                    }
                })
        .next("middle").subtype(String.class).where(
                new SimpleCondition<String>() {
                    private static final long serialVersionUID = 1L;

                    @Override
                    public boolean filter(String value) throws Exception {
                        return value.endsWith("er");
                    }
                })
        .followedBy("end").where(
                new SimpleCondition<String>() {
                    private static final long serialVersionUID = 1L;

                    @Override
                    public boolean filter(String value) throws Exception {
                        return value.length() > 9;
                    }
                });
        
        PatternStream<String> patternStream = CEP.pattern(inputStream, pattern);
        SingleOutputStreamOperator<String> outStream = patternStream.select(new PatternSelectFunction<String, String>() {
            @Override
            public String select(Map<String, List<String>> pattern) throws Exception {
                return null;
            }
        });
    }
    
    public static class LogMap extends ProcessWindowFunction<PeterEntity, Tuple4<String, AtomicLong, AtomicLong, AtomicLong>, Tuple, TimeWindow> {
        private static final long serialVersionUID = 1L;
        private ListState<PeterEntity> list = null;
        
        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            ListStateDescriptor<PeterEntity> descriptor = new ListStateDescriptor<PeterEntity>("list", PeterEntity.class);
            list = getRuntimeContext().getListState(descriptor);
        }

        @Override
        public void process(Tuple key,
            ProcessWindowFunction<PeterEntity, Tuple4<String, AtomicLong, AtomicLong, AtomicLong>, Tuple, TimeWindow>.Context context,
            Iterable<PeterEntity> elements,
            Collector<Tuple4<String, AtomicLong, AtomicLong, AtomicLong>> out) throws Exception
        {
            Tuple4<String, AtomicLong, AtomicLong, AtomicLong> tuple =new Tuple4<String, AtomicLong,
                AtomicLong, AtomicLong>("", new AtomicLong(0), new AtomicLong(Long.MAX_VALUE), new AtomicLong(0));
            Set<String> set = new HashSet<String>();
            Log.info(">>>> process: " + context.currentWatermark());

            for (PeterEntity element : elements) {
                set.add(element.f1);
                tuple.f1.incrementAndGet();

                if (tuple.f2.get() > element.f0) {
                    tuple.f2.set(element.f0);
                }

                if (tuple.f3.get() < element.f0) {
                    tuple.f3.set(element.f0);
                }
            }

            tuple.f0 = set.toString();
            out.collect(tuple);
        }
    }
    
    class PeterTestSink extends RichSinkFunction<TaskDTO> {
        private static final long serialVersionUID = 1L;

        @Override
        public void close() throws Exception {
            super.close();
        }
        
        @Override
        public void invoke(TaskDTO value, Context context) throws Exception {
            super.invoke(value, context);
        }
    }
}
